/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */



package com.bff.gaia.unified.sdk.transforms.xjoin.operators;



import java.util.ArrayList;

import java.util.List;



import com.bff.gaia.unified.sdk.coders.RowCoder;

import com.bff.gaia.unified.sdk.schemas.Schema;

import com.bff.gaia.unified.sdk.transforms.ParDo;

import com.bff.gaia.unified.sdk.transforms.xjoin.core.enums.ECacheType;

import com.bff.gaia.unified.sdk.transforms.xjoin.core.enums.EJoinType;

import com.bff.gaia.unified.sdk.transforms.xjoin.core.side.FieldInfo;

import com.bff.gaia.unified.sdk.transforms.xjoin.core.side.JoinInfo;

import com.bff.gaia.unified.sdk.transforms.xjoin.core.side.SideTableInfo;

import com.bff.gaia.unified.sdk.values.PCollection;

import com.bff.gaia.unified.sdk.values.Row;



public class Xjoin {



	private PCollection<Row> left;

	private SideTableInfo right;

	private List<FieldInfo> outFieldInfoList; // 输出列

	private JoinInfo joinInfo; // 连接辅助参数



	public Xjoin(PCollection<Row> left, SideTableInfo right) {

		this.left = left;

		this.right = right;

	}



	public ProjectLeft projectLeftKey(int... leftKeyPosition) {

		return new ProjectLeft(leftKeyPosition);

	}



	public class ProjectLeft {

		int[] leftKeyPosition;

		int[] leftFieldPosition;

		List<Schema.Field> leftKeyType;



		ProjectLeft(int... leftKeyPosition) {



			this.leftKeyPosition = leftKeyPosition;

			//			System.out.println(left.getType().getArity());

			this.leftKeyType = projectFields(((RowCoder) left.getCoder()).getSchema().getFields(), leftKeyPosition);

			leftFieldPosition = projectPosition(((RowCoder) left.getCoder()).getSchema().getFields().size());

		}



		//		ProjectLeft(String... leftKeyName) {

		//			int[] tempKey = new int[leftKeyName.length];

		//			for (int i = 0; i < tempKey.length; i++) {

		//				tempKey[i] = ((RowTypeInfo)left.getType()).getFieldIndex(leftKeyName[i]);

		//			}

		//			this.leftKeyPosition = tempKey;

		//

		//			this.leftKeyType = RowTypeInfo.projectFields((RowTypeInfo)left.getType(),leftKeyPosition);

		//

		//			leftFieldPosition = projectPosition(left.getType().getArity());

		//		}



		public ProjectLeft projectLeftField(int... leftFieldPosition) {



			if (isSubSets(leftFieldPosition, leftKeyPosition)) {

				throw new IllegalArgumentException("project fields must have key");

			}



			this.leftFieldPosition = leftFieldPosition;



			return this;

		}



		public ProjectRight projectRightKey(int... rightKeyPosition) {

			if (leftKeyPosition.length != rightKeyPosition.length) {

				throw new IllegalArgumentException("wrong key num");

			}

			return new ProjectRight(rightKeyPosition);

		}



		//		public ProjectRight projectRightKey(String... rightKeyName) {

		//			if (leftKeyPosition.length != rightKeyName.length) {

		//				throw new IllegalArgumentException("wrong key num");

		//			}

		//			return new ProjectRight(rightKeyName);

		//		}



		public class ProjectRight {

			int[] rightKeyPosition;

			int[] rightFieldPosition;

			List<Schema.Field> rightKeyType;



			ProjectRight(int... rightKeyPosition) {

				this.rightKeyPosition = rightKeyPosition;



				rightKeyType = projectFields(right.getFieldTypeList(), rightKeyPosition);



				rightFieldPosition = projectPosition(right.getFieldList().size());

			}



			public ProjectRight projectRightField(int... rightFieldPosition) {

				if (isSubSets(rightFieldPosition, rightKeyPosition)) {

					throw new IllegalArgumentException("project fields must have key");

				}

				this.rightFieldPosition = rightFieldPosition;

				return this;

			}



			public SetArgs setCacheType(ECacheType cacheType) {

				return new SetArgs(cacheType);

			}



			public class SetArgs {

				private String cacheType = "LRU"; // None or LRU or ALL



				private int cacheSize = 10000;



				private long cacheTimeout = 60 * 1000; //



				private int asyncCapacity = 100;



				SetArgs(ECacheType cacheType) {

					this.cacheType = cacheType.name();

				}



				public SetArgs setCacheSize(int cacheSize) {

					this.cacheSize = cacheSize;

					return this;

				}



				public SetArgs setCacheTimeout(long cacheTimeout) {

					this.cacheTimeout = cacheTimeout;

					return this;

				}



				public SetArgs setAsyncCapacity(int asyncCapacity) {

					this.asyncCapacity = asyncCapacity;

					return this;

				}



				public PCollection<Row> apply() {

					// 构建缓存信息

					right.setCacheType(cacheType);

					right.setCacheSize(cacheSize);

					right.setCacheTimeout(cacheTimeout);

					right.setAsyncCapacity(asyncCapacity);



					ArrayList<Schema.Field> fieldArrayList = new ArrayList<>();

					ArrayList<String> strings = new ArrayList<>();



					// 构建输出数据结构

					outFieldInfoList = new ArrayList<>();

					for (int i : leftFieldPosition) {

						FieldInfo fieldInfo = new FieldInfo(((RowCoder) left.getCoder()).getSchema().getField(i));

						for (int ii : leftKeyPosition) {

							if (i == ii) {

								fieldInfo.setJoinKey(true);

								break;

							}

						}

						fieldInfo.setTable("stream");

						outFieldInfoList.add(fieldInfo);



						fieldArrayList.add(fieldInfo);

						strings.add(fieldInfo.getName());

					}

					for (int j : rightFieldPosition) {

						FieldInfo fieldInfo = new FieldInfo(right.getFieldTypeList().get(j));

						for (int jj : rightKeyPosition) {

							if (j == jj) {

								fieldInfo.setJoinKey(true);

								break;

							}

						}

						fieldInfo.setTable(right.getName());

						if (!fieldInfo.isJoinKey()) {

							fieldArrayList.add(fieldInfo);

							strings.add(fieldInfo.getName());

						}

						outFieldInfoList.add(fieldInfo);

					}



					// 构建连接信息

					JoinInfo joinInfo = new JoinInfo();

					joinInfo.setLeftTableName("stream");

					joinInfo.setLeftTableAlias("stream");

					joinInfo.setLeftIsSideTable(false);

					joinInfo.setRightTableName(right.getName());

					joinInfo.setRightTableAlias(right.getName());

					joinInfo.setRightIsSideTable(true);

					joinInfo.setJoinType(EJoinType.EQUAL);



					// 选择缓存模式

					return

							left.apply("xjoin with "+right.getName(),

									ParDo.of(

											XjoinAllOperator.getSideJoinDataStream(

													right.getType(),

													((RowCoder) left.getCoder()).getSchema(),

													joinInfo,

													outFieldInfoList,

													right)))

									.setCoder(RowCoder.of(new Schema(fieldArrayList)));

				}

			}

		}

	}



	//	public static List<Schema.Field> projectSchema(Schema schema,int[] fieldMapping){

	//		List<Schema.Field> fieldList=new ArrayList<>();

	//		for (int value : fieldMapping) {

	//			fieldList.add(schema.getField(value));

	//		}

	//		return fieldList;

	//	}



	public static List<Schema.Field> projectFields(List<Schema.Field> typeInfo, int[] fieldMapping) {

		List<Schema.Field> fieldList = new ArrayList<>();

		for (int value : fieldMapping) {

			fieldList.add(typeInfo.get(value));

		}

		return fieldList;

	}



	private static int[] projectPosition(int l) {

		int[] temp = new int[l];

		for (int i = 0; i < l; i++) {

			temp[i] = i;

		}

		return temp;

	}



	private static boolean isSubSets(int[] father, int[] child) {

		for (int i : child) {

			boolean isSubSet = false;

			for (int j : father) {

				if (j == i) {

					isSubSet = true;

					break;

				}

			}

			if (!isSubSet) {

				return true;

			}

		}

		return false;

	}

}